package com.lmq

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType, StructField, StructType}

import scala.collection.mutable

/**
 * TODO: not used .
 *        我们在程序中，
 *        划分2种信息，
 *        类别和价格。
 *        类别 包含了1 - 12这12个数字 以及S整个字符
 *        价格则是8到10位的数字，因此我们根据上述特征，将价格和特征区分开。
 *      上述描述错误,下面给出纠正：
 *      yoochoose-clicks只给出了category信息，有3种方式表示它的信息
 *       1 1-12这些数字
 *      2 S 表示special offer
 *      3 8-12位数字
 *   该注释描述的内容正确，但是下面的代码不再使用。
 */
object ItemInforDistince {




  Logger.getLogger("org.apache.spark")
    .setLevel(Level.WARN)

  val spark = SparkSession.builder()
    .master("local[*]")
    .appName("Test")
    .getOrCreate()

  def distCat: UserDefinedFunction = udf((x:String)=>  {
    if (x.length>=8 && x.length<=10 ){
      2
    } else 1
  } )


//  val  arrtoStrFnc: UserDefinedFunction = udf((x:mutable.WrappedArray[mutable.WrappedArray[AnyVal]])=> {
//  val  arrtoStrFnc: UserDefinedFunction = udf((x:mutable.WrappedArray[mutable.WrappedArray[AnyVal]])=> {
//    var strings = ""
//    if (x.length==2){
//      if( x(0)(0).toString.length<8){
//        var categ =x(0).mkString("|")
//        categ=categ+","
//      }
//      else {
//        price
//      }
//    }
////    for (  i <- x.indices){
//////      if( x.length==1)
////
////
////
////      strings=strings+x(i).mkString("|")
////      if( i != x.length-1){ strings=strings+"," }
////    }
//    strings
//
//
//  })

//  val timeTransfer: UserDefinedFunction = udf((x:String)=>  utils.timeStr2Tsp(x) )

  def main(args: Array[String]): Unit = {
    val utils = new utils()
    //csv sechema
    val schema2 = StructType(
      Array(

        StructField("ItemId", IntegerType, nullable = true),
        StructField("Context", StringType, nullable = true)

      )
    )


    // read CSV
    val vc:DataFrame = spark.read
      .schema(schema2)
      .csv("file:///home/iptv/yoochoose/OneOsixfour_sessionsRevisev4/part-00000-c060d1e9-2dc1-439c-9238-6eff595c5744-c000.csv")


    vc.printSchema()



    println("=====================================")

    def sortwithPositionElement(Pt:Int) = udf((x:mutable.WrappedArray[Double])=>  x(Pt) )
    //      println(v.count())

    vc.createTempView("vc")
    val toStrFnc = udf((x:mutable.WrappedArray[AnyVal])=>x.mkString("|"))
    spark.udf.register("distCat",distCat)
    spark.udf.register("toStrFnc",toStrFnc)
//    spark.udf.register("arrtoStrFnc",arrtoStrFnc)

//
//    println("alllines "+spark.sql(
//      """
//        |select ItemId,collect_list(Context)
//        |from
//        |(select ItemId, Context, distCat(vc.Context) as Sepr
//        |from vc) vp
//        |group by ItemId,Sepr
//        |""".stripMargin)
//      .count())
//    spark.sql(
//      """
//        |select count(distinct(ItemId)) as sumofitemid
//        |from
//        |(
//        |select ItemId
//        |from
//        |(select ItemId, Context, distCat(vc.Context) as Sepr
//        |from vc) vp
//        |group by ItemId,Sepr) AP
//        |""".stripMargin).show(1000,false)

//     spark.sql(
//      """
//        |select ItemId,collect_list(Context) as Context
//        |from
//        |(select ItemId, Context, distCat(vc.Context) as Sepr
//        |from vc) vp
//        |group by ItemId,Sepr
//        |""".stripMargin).createTempView("tmp")

    val w = spark.sql(
      """
        |select ItemId, toStrFnc(collect_list(Context))
        |from vc
        |group by ItemId
        |""".stripMargin)
//      .show(false)
//    spark.sql(
//      """
//        |select  ItemId, arrtoStrFnc(collect_list(Context))
//        |from tmp
//        |group by ItemId
//        |""".stripMargin)
////
//    spark.sql(
//      """
//        |select  ItemId, collect_list(Context)
//        |from tmp
//        |group by ItemId
//        |""".stripMargin)
//      .show(false)
//      .coalesce(1)
//      .write.option("header","false")
//      .csv("file:///home/iptv/yoochoose/OneOsixfour_itemInfo.csv")


//       .show(10000,false)


    w.distinct()
      .coalesce(1).write
      .option("header","false")
//      final item information is located in this directory.
      .csv("file:///home/iptv/yoochoose/OneOsixfour_itemv2")
    println("Done.")





  }

}
